package data.platform.merge.internal.query;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import data.platform.common.merge.AggregateItem;
import data.platform.common.merge.ArithmeticType;
import data.platform.common.merge.CommonAggregateItem;
import data.platform.common.query.QueryMetric;
import data.platform.common.service.command.KeyValueRepository;
import data.platform.common.service.query.DataAggregateQueryService;
import data.platform.common.service.query.MetricResultQueryService;
import data.platform.common.service.query.MetricTagQueryService;
import data.platform.merge.internal.cache.AggregateItemCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
@RequiredArgsConstructor
public class DataAggregateQueryServiceImpl implements DataAggregateQueryService {

    final KeyValueRepository<String, byte[]> keyValueRepository;

    final MetricTagQueryService metricTagQueryService;

    final MetricResultQueryService metricResultQueryService;

    final AggregateItemCache aggregateItemCache;

    private ObjectMapper objectMapper;

    @PostConstruct
    public void init() {
        objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
    }

    public Flux<Tuple2<String, AggregateItem>> queryAggregateItems(ArithmeticType arithmeticType, String metric, Map<String, Set<String>> tags) {

        // 要查询的标签
        Set<Integer> tagIds;
        if (Objects.isNull(tags) || tags.size() == 0) {
            // 只通过指标查询
            tagIds = getTagIds(metricTagQueryService.matchingTagByMetric(metric));
        } else {
            Set<String> tagJsons = new HashSet<>();
            // 获取需要查询的指标以及对应的标签
            QueryMetric queryMetric = new QueryMetric();
            queryMetric.setTags(tags);
            List<Map<String, String>> queryTags = metricResultQueryService.getMetricTags(queryMetric);
            for (Map<String, String> tagMap : queryTags) {
                tagJsons.addAll(metricTagQueryService.matchingTag(metric, tagMap));
            }
            tagIds = getTagIds(tagJsons);
        }

        return Flux.fromIterable(tagIds)
                .flatMap(tagId -> aggregateItemCache.find(arithmeticType, metric, tagId)
                        .filter(aggregateItem -> Objects.nonNull(aggregateItem))
                        .map(aggregateItem -> Tuples.of(metricTagQueryService.getTagJson(tagId).get(), aggregateItem)));
    }

    private Set<Integer> getTagIds(Collection<String> tagJsons) {
        return tagJsons
                .stream()
                .map(tagJson -> metricTagQueryService.findTagId(tagJson))
                .filter(Optional::isPresent)
                .map(Optional::get)
                .collect(Collectors.toSet());
    }
}
